package com.yyq.bigexport.util.insert;

import cn.hutool.core.thread.NamedThreadFactory;
import com.yyq.bigexport.entity.User;
import com.yyq.bigexport.service.UserService;
import com.yyq.bigexport.util.UserGenerator;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 *
 * @author hsg
 * @date 2024/1/11 15:21
 */
@Slf4j
public class InsertTask implements Runnable {
    private final UserService userService;
    //每次批量插入总数量
    private int batchSize;
    //每次批量单纯插入数量10000
    private final static int insertSize = 2000;
    private  final CountDownLatch countDownLatch;
    ThreadPoolExecutor executorService = new ThreadPoolExecutor(50,
            51,
            30,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new NamedThreadFactory("insert-pool-", false));
    public InsertTask(UserService userService, int batchSize, CountDownLatch countDownLatch) {
        this.userService = userService;
        this.batchSize = batchSize;
        this.countDownLatch = countDownLatch;
    }
    private CountDownLatch generationLatch = new CountDownLatch(1);
    @Override
    public void run() {
        log.info("插入数据线程开始 线程名={}",Thread.currentThread().getName());
        try {
            //并发生成用户数
            List<User> records = UserGenerator.generateUsersAndPrintStats(batchSize);
            log.info("线程{}生成数据，总数={}", Thread.currentThread().getName(), records.size());
            // 通知主线程本线程生成数据完成
            generationLatch.countDown();

            // 等待所有线程都完成数据生成
            generationLatch.await();
            batchInsert(records);
            // 清空allUsers
            records.clear();
            // 通知gc回收内存
            System.gc();
        } catch (Exception e){
            log.error("插入数据失败", e);
        } finally{
            // 清空数据
            //计数器减一
            countDownLatch.countDown();
            // 可以添加一些资源释放等操作
            log.info("线程{}释放资源", Thread.currentThread().getName());
        }
    }


    public void batchInsert(List<User> records) {

        String threadName = Thread.currentThread().getName();
        log.info("线程{}批量插入数据，总数={}", threadName, records.size());
        int size = records.size();
        int batchCount = (size + insertSize - 1) / insertSize;
        log.info("线程={}插入数据总批次={}", threadName, batchCount);
        int insertErrCount = 0;

        List<Future<Void>> futures = new ArrayList<>();
        for (int i = 0; i < batchCount; i++) {
            final int finalI = i;
            Future<Void> future = executorService.submit(() -> {
                try {
                    String thisThreadName = threadName + "_" + Thread.currentThread().getName()+"批次——"+finalI;
                    int endIndex = Math.min((finalI + 1) * insertSize, size);
                    List<User> batch = records.subList(finalI * insertSize, endIndex);
                    log.info("线程{}正在插入数据{}条", thisThreadName, batch.size());
                    userService.saveBatch(batch, insertSize);
                    log.info("线程{}插入完成", thisThreadName);
                } catch (Exception e) {
                    log.error("批量插入失败 err={}", e.getMessage(), e);
                }
                return null;
            });
            futures.add(future);
        }

        // 等待所有任务完成
        for (Future<Void> future : futures) {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                log.error("等待插入任务完成时发生异常", e);
            }
        }

        log.info("线程={}，插入数据完成，总数={}, 成功数={}, 失败数={}", Thread.currentThread().getName(), size, size - insertErrCount, insertErrCount);
    }


}
